package cn.monkey.io.transport.mq.rabbitmq;

import cn.monkey.io.transport.mq.MqInboundConfig;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.function.Predicate;
import java.util.function.Supplier;

public class SimpleRabbitMqInbound implements RabbitMqInbound {

    private final RabbitMqInboundConfig inboundConfig;
    protected final RabbitMqFluxReceive receiver;
    protected final Collection<Channel> channels;
    protected final Supplier<Mono<Void>> closeAction;

    public SimpleRabbitMqInbound(Collection<Channel> channels,
                                 RabbitMqInboundConfig inboundConfig,
                                 Supplier<Mono<Void>> closeAction) {
        this.inboundConfig = inboundConfig;
        this.channels = channels;
        this.closeAction = closeAction;
        this.receiver = new RabbitMqFluxReceive(channels, inboundConfig.getTopics());
    }

    @Override
    public MqInboundConfig inboundConfig() {
        return inboundConfig;
    }

    @Override
    public Mono<Void> close() {
        this.receiver.dispose();
        return this.closeAction.get();
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver
                .map(byte[].class::cast)
                .map(bytes -> this.inboundConfig.getDecoder().decode(bytes))
                .filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }
}
